package com.ricky.mq.client.connection;

import cn.hutool.core.util.HexUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.ricky.mq.client.code.constant.MessageTemplate;
import com.ricky.mq.client.code.handler.ServerDataHandler;
import com.ricky.mq.client.code.manager.HeartbeatCheckManager;
import com.ricky.mq.client.session.AsyncBoringSession;
import com.ricky.mq.client.session.BoringSession;
import com.ricky.mq.client.session.SyncBoringSession;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 客户端连接器
 *
 * @author ricky
 * @since 2020.06.18
 */
public class ClientConnection implements Runnable {
    /**
     * 记日志的
     */
    private static Log LOG = LogFactory.get(ClientConnection.class);
    /**
     * 本身
     */
    private static ClientConnection clientConnection = new ClientConnection();
    /**
     * 心跳检查定时器
     */
    private ScheduledExecutorService scheduledExecutorService;
    /**
     * 会话管理器
     */
    private BoringSession boringSession;
    /**
     * 连接通道
     */
    private SocketChannel socketChannel;
    /**
     * 是否与服务器断开连接，默认否
     */
    private volatile boolean isDisconnected = false;
    /**
     * IP
     */
    private String ip = "127.0.0.1";
    /**
     * 端口
     */
    private int port = 2428;
    /**
     * 心跳检查时间，单位秒
     */
    private int heartbeatCheckTime = 10;
    /**
     * 数据等待超时时间,单位毫秒
     */
    private long waitTimeOut = 5000L;
    /**
     * 轮训器
     */
    private Selector selector;
    /**
     * 读buff
     */
    private ByteBuffer readBuf = ByteBuffer.allocate(0xffff);
    /**
     * 写buff
     */
    private ByteBuffer writeBuf;

    private ClientConnection() {
        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleWithFixedDelay(new HeartbeatCheckManager(), heartbeatCheckTime, heartbeatCheckTime, TimeUnit.SECONDS);
    }

    /**
     * 设置ip、端口号,默认ip本机，默认断开2428
     *
     * @param ip   ip
     * @param port 端口
     * @return ClientConnection
     */
    public ClientConnection ipAndPort(String ip, int port) {
        this.ip = ip;
        this.port = port;
        return this;
    }

    /**
     * 设置发送模式，默认同步发送
     *
     * @param model BoringSession.ASYNC_MODEL 异步；BoringSession.SYNC_MODEL 同步。
     * @return ClientConnection
     */
    public ClientConnection sendModel(int model) {
        switch (model) {
            case BoringSession.ASYNC_MODEL:
                this.boringSession = new AsyncBoringSession();
                break;
            case BoringSession.SYNC_MODEL:
                this.boringSession = new SyncBoringSession();
            default:
                break;

        }
        return this;
    }

    /**
     * 设置读buff大小，默认65535
     *
     * @param readBufSize
     * @return ClientConnection
     */
    public ClientConnection readBufSize(int readBufSize) {
        readBuf = ByteBuffer.allocate(readBufSize);
        return this;
    }

    /**
     * 设置心跳检查时间，默认10秒
     *
     * @param heartbeatCheckTime 心跳检查时间，单位秒
     * @return ClientConnection
     */
    public ClientConnection heartbeatCheckTime(int heartbeatCheckTime) {
        this.heartbeatCheckTime = heartbeatCheckTime;
        return this;
    }

    /**
     * 设置数据等待超时时间，默认5000毫秒
     *
     * @param waitTimeOut
     * @return ClientConnection
     */
    public ClientConnection waitTimeOut(long waitTimeOut) {
        this.waitTimeOut = waitTimeOut;
        return this;
    }

    /**
     * 开启连接
     *
     * @return ClientConnection
     */
    public ClientConnection connection() {
        try {
            socketChannel = SocketChannel.open();
            socketChannel.connect(new InetSocketAddress(ip, port));
            socketChannel.configureBlocking(false);
            selector = Selector.open();
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            isDisconnected = false;
            start();
        } catch (IOException e) {
            try {
                socketChannel.close();
            } catch (IOException ioException) {

            }
            isDisconnected = true;
            LOG.error("BoringMQ-Client 启动失败,MQ服务器未连接成功！" + heartbeatCheckTime + "秒钟之后将重试！", e);
        }
        return this;
    }

    /**
     * 启动client端
     *
     * @return
     */
    public void start() {
        new Thread(this).start();
    }

    /**
     * 接收消息核心业务
     */
    @Override
    public void run() {
        try {
            while (true) {
                selector.select();
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    switch (key.readyOps()) {
                        case SelectionKey.OP_READ:
                            SocketChannel readChannel = (SocketChannel) key.channel();
                            readBuf.clear();
                            int size = 0;
                            try {
                                size = readChannel.read(readBuf);
                            } catch (IOException e) {
                                closeSocket(readChannel);
                                break;
                            }
                            if (size >= 0) {
                                readBuf.flip();
                                byte[] array = new byte[readBuf.remaining()];
                                readBuf.get(array);
                                ServerDataHandler.handler(readChannel, new String(array));
                                readBuf.clear();
                                key.interestOps(SelectionKey.OP_READ);
                            } else {
                                closeSocket(readChannel);
                            }
                            break;
                        default:
                            break;
                    }
                    iterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 关闭socketChannel通道
     *
     * @param socketChannel
     */
    private void closeSocket(SocketChannel socketChannel) {
        try {
            if (ObjectUtil.isNotEmpty(socketChannel)) {
                LOG.info("服务器连接丢失！" + socketChannel.getRemoteAddress());
                isDisconnected = true;
                socketChannel.socket().close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 同步发送数据
     *
     * @param commandCode 命令码
     * @param data        消息对象
     * @return
     */
    public boolean sendData(String commandCode, Object data) {
        String assemblyData = MessageTemplate
                .build()
                .setCommandCode(commandCode)
                .setData(HexUtil.encodeHexStr(JSONUtil.parseObj(data).toString()))
                .assemblyData();
        try {
            if (socketChannel.isOpen()) {
                writeBuf = ByteBuffer.wrap(assemblyData.getBytes());
                socketChannel.write(writeBuf);
                writeBuf.clear();
                socketChannel.register(selector, SelectionKey.OP_READ);
                return true;
            }
            return false;
        } catch (Exception e) {
            LOG.error("发送消息失败！", e);
            return false;
        }
    }

    /**
     * 获取会话，如果没有设置同步或异步，默认同步会话
     *
     * @return BoringSession
     */
    public BoringSession getBoringSession() {
        if (ObjectUtil.isNull(boringSession)) {
            SyncBoringSession syncBoringSession = new SyncBoringSession();
            this.boringSession = syncBoringSession;
        }
        return this.boringSession;
    }

    /**
     * 获取客户端连接管理器
     *
     * @return ClientConnection
     */
    public static ClientConnection getClientConnection() {
        return clientConnection;
    }

    /**
     * 返回连接是否断开
     *
     * @return isDisconnected
     */
    public boolean isDisconnected() {
        return isDisconnected;
    }

    /**
     * 返回心跳检查时间
     * @return 心跳检查时间
     */
    public int getHeartbeatCheckTime() {
        return heartbeatCheckTime;
    }

    /**
     * 返回心跳超时次数
     * @return 心跳超时次数
     */
    public long getWaitTimeOut() {
        return this.waitTimeOut;
    }
}
